package mongodb

import (
	"fg-engine/queue"
	"fmt"

	"gopkg.in/mgo.v2"
	"gopkg.in/mgo.v2/bson"
)

const (
	// 操作类型
	ID_ACT_INSERT = 1
	ID_ACT_UPDATE = 2
	ID_ACT_UPSERT = 3
	ID_ACT_REMOVE = 4
	ID_ACT_QUERY  = 5

	ID_PROC_READ_NUM  = 16 // 读协程数量
	ID_PROC_WRITE_NUM = 16 // 写协程数量

	ID_QUEUE_LEN = 20480
)

type QueueType chan *DBExecInfo

var (
	writeQueue   QueueType
	readQqueue   QueueType
	resultQueue  QueueType
	resultArray1 *queue.ArrayQueue
	resultArray2 *queue.ArrayQueue
)

func Run() {
	writeQueue = make(QueueType, ID_QUEUE_LEN)
	readQqueue = make(QueueType, ID_QUEUE_LEN)
	resultQueue = make(QueueType, ID_QUEUE_LEN)
	resultArray1 = queue.NewArrayQueue(ID_QUEUE_LEN)
	resultArray2 = queue.NewArrayQueue(ID_QUEUE_LEN)

	for i := 0; i < ID_PROC_READ_NUM; i++ {
		go processRead()
	}

	for i := 0; i < ID_PROC_WRITE_NUM; i++ {
		go processWrite()
	}

	go prcessResult()
}

func PushRead(info *DBExecInfo) bool {
	select {
	case readQqueue <- info:
	default:
		fmt.Println("db pushRead full!")
		return false
	}
	return true
}

func PushWrite(info *DBExecInfo) bool {
	select {
	case writeQueue <- info:
	default:
		fmt.Println("db pushWrite full!")
		return false
	}
	return true
}

// 逻辑主线程调用
func Loop() {
	// for {
	// 	select {
	// 	case info := <-resultQueue:
	// 		if info.CallFunc != nil {
	// 			info.CallFunc(info)
	// 		}
	// 		PutDBExecInfo(info)
	// 	default:
	// 		return
	// 	}
	// }
	for {
		if i, ok := resultArray1.Read(); ok {
			info := i.(*DBExecInfo)
			if info.CallFunc != nil {
				info.CallFunc(info)
			}
			resultArray2.Write(info)
		} else {
			break
		}
	}
}

// 回调消息收集
func prcessResult() {
	for {
		select {
		case info := <-resultQueue:
			resultArray1.Write(info)
			for {
				if i, ok := resultArray2.Read(); ok {
					d := i.(*DBExecInfo)
					PutDBExecInfo(d)
				} else {
					break
				}
			}
		}
	}
}

func processRead() {
	session := GetSession()
	expectdFields := bson.M{"_id": 0}

	for {
		info := <-readQqueue
		if info.Collection == "" {
			fmt.Println("processRead Collection empty!")
			continue
		}

		for {
			ret := func() bool {
				defer func() {
					if err := recover(); err != nil {
						fmt.Printf("processWrite act %d, coll %s err %s\n", info.ActType, info.Collection, err)
					}
				}()

				c := session.DB("").C(info.Collection)
				switch info.ActType {
				case ID_ACT_QUERY:
					if info.Skip == 0 && info.Limit == 0 {
						info.Err = c.Find(info.Cond).Select(expectdFields).All(&info.Result)
					} else {
						info.Err = c.Find(info.Cond).Skip(info.Skip).Limit(info.Limit).Select(expectdFields).All(&info.Result)
					}
				default:
					fmt.Printf("processRead unknow actiontype %d, %s \n", info.ActType, info.Collection)
					return false
				}
				return true
			}()

			if !ret {
				break
			}

			if info.Err != nil && !mgo.IsDup(info.Err) && info.Err != mgo.ErrNotFound && info.Err != mgo.ErrCursor {
				fmt.Printf("processRead Reconnect act %d, coll %s, err %s\n", info.ActType, info.Collection, info.Err.Error())
				session.Close()
				session = Reconnect()
				continue
			}

			if info.CallFunc != nil {
				resultQueue <- info
			} else {
				PutDBExecInfo(info)
			}
			break
		}
	}
	session.Close()
}

func processWrite() {
	session := GetSession()
	for {
		info := <-writeQueue
		if info.Collection == "" {
			fmt.Println("processWrite Collection empty!")
			continue
		}

		for {
			ret := func() bool {
				defer func() {
					if err := recover(); err != nil {
						fmt.Printf("processWrite act %d, coll %s err %s\n", info.ActType, info.Collection, err)
					}
				}()

				c := session.DB("").C(info.Collection)
				switch info.ActType {
				case ID_ACT_INSERT:
					info.Err = c.Insert(info.Doc)
				case ID_ACT_UPDATE:
					info.Err = c.Update(info.Cond, bson.M{"$set": info.Doc})
				case ID_ACT_UPSERT:
					_, info.Err = c.Upsert(info.Cond, bson.M{"$set": info.Doc})
				case ID_ACT_REMOVE:
					info.Err = c.Remove(info.Cond)
				default:
					fmt.Printf("processWrite unknow actiontype %d, %s \n", info.ActType, info.Collection)
					return false
				}
				return true
			}()

			if !ret {
				break
			}

			if info.Err != nil && !mgo.IsDup(info.Err) && info.Err != mgo.ErrNotFound && info.Err != mgo.ErrCursor {
				fmt.Printf("processWrite Reconnect act %d, coll %s, err %s\n", info.ActType, info.Collection, info.Err.Error())
				session.Close()
				session = Reconnect()
				continue
			}

			if info.CallFunc != nil {
				resultQueue <- info
			} else {
				PutDBExecInfo(info)
			}
			break
		}
	}
	session.Close()
}
